#include <iostream>
#include <vector>
#include <mutex>
#include <condition_variable>
#include <thread>
#include <functional>
#include <memory>
#include <unistd.h>

class EventLoop {
    private:
        using Functor = std::function<void()>;
        std::thread::id _thread_id;
        /*因为有可能当前eventloop线程卡在等待描述符就绪事件上，因此通过给weakupfd写入数据唤醒阻塞*/
        int _weakup_fd;
        int _timeout;
        Poller *_poller;
        Channel *_weakup_channel;
        TimerQueue *_timer_queue;

        std::mutex _mutex;
        std::vector<Functor> _pending_functor;
    private:
        static int EventLoop::create_event_fd() {
            int fd = ::eventfd(0, EFD_CLOEXEC | EFD_NONBLOCK);
            assert(fd != -1);
            return fd;
        }
        void EventLoop::weakup() {
            uint64_t opts = 1;
            int ret = ::write(_weakup_fd, &opts, sizeof(opts));
            if (ret != sizeof(opts)) {
                LOG("write weakup failed!! %d-%d-%s", errno, EBADF, strerror(errno));
                abort();
            }
        }
        void EventLoop::weakup_channel_handler_read() {
            uint64_t opts;
            int ret = ::read(_weakup_fd, &opts, sizeof(opts));
            if (ret != sizeof(opts)) {
                LOG("read weakup failed!!");
                abort();
            }
        }
        void EventLoop::do_pending_functor() {
            std::vector<Functor> functor;
            {
                std::unique_lock<std::mutex> lock(_mutex);
                functor.swap(_pending_functor);
            }
            for (const Functor &f : functor) {
                f();
            }
        }
    public:
        EventLoop::EventLoop():
            _thread_id(std::this_thread::get_id()),
            _weakup_fd(EventLoop::create_event_fd()),
            _poller(new Poller(this)),
            _weakup_channel(new Channel(this, _weakup_fd)),
            _timer_queue(new TimerQueue(this)){
            _weakup_channel->set_read_callback(std::bind(&EventLoop::weakup_channel_handler_read, this));
            _weakup_channel->enable_read();
        }
        void EventLoop::start() {
            LOG("EventLoop start...");
            while(1) {
                std::vector<Channel *> active_channels;
                bool ret = _poller->poll(&active_channels, DEFAULT_TIMEOUT * 1000);
                if (ret == false) {
                    usleep(1000);
                    continue;
                }
                for (Channel *channel : active_channels) {
                    channel->handle_event();
                }
                do_pending_functor();
            }
            return ;
        }
        /*如果当前就是在eventloop线程中，则直接调用执行，否则就压入到任务队列中*/
        void EventLoop::run_in_loop(Functor cb) {
            if (is_in_loop_thread()) {
                return cb();
            }
            return queue_in_loop(cb);
        }
        /*将需要执行的函数压入eventloop线程任务队列中，然后如果当前不在eventloop线程中，则唤醒eventloop线程*/
        void EventLoop::queue_in_loop(Functor cb) {
            {
                std::unique_lock<std::mutex> lock(_mutex);
                _pending_functor.push_back(cb);
            }
            if (is_in_loop_thread() == false) {
                weakup();
            }
            return ;
        }
        bool EventLoop::is_in_loop_thread() {
            return _thread_id == std::this_thread::get_id();
        }
        void EventLoop::update_channel(Channel *channel) {
            return _poller->update_channel(channel);
        }
        void EventLoop::remove_channel(Channel *channel) {
            return _poller->remove_channel(channel);
        }
        bool EventLoop::has_channel(Channel *channel) {
            return _poller->has_channel(channel);
        }
        void EventLoop::timer_task_enqueue(const OnTimerCallback &cb, uint64_t timer_id) {
            return _timer_queue->timer_add(cb, timer_id);
        }
        void EventLoop::timer_refresh(uint64_t timer_id) {
            return _timer_queue->timer_refresh(timer_id);
        }
};

/*把eventloop和thread放一起当作一个整体，保证在线程内实例化eventloop才能确保eventloop的thread_id成员是对应线程ID，run_in_loop才更靠谱*/
class LoopThread {
    private:
        EventLoop *_loop;
        std::mutex _mutex;
        std::condition_variable _cond;
        std::thread _thread;
    private:
        void LoopThread::thread_entry() {
            /*在线程内部实例化eventloop，这样才能设置好eventloop对应的线程id*/
            EventLoop loop;
            {
                std::unique_lock<std::mutex> lock(_mutex);
                _loop = &loop;
                _cond.notify_all();
            }
            return loop.start();
        }
    public:
        LoopThread::LoopThread():_loop(NULL), _thread(std::thread(&LoopThread::thread_entry, this)) {}
        /*如果外界想要获取线程对应的eventloop，则阻塞至_loop不为NULL*/
        EventLoop *LoopThread::own_loop() { 
            EventLoop *loop = NULL;
            {
                std::unique_lock<std::mutex> lock(_mutex);
                _cond.wait(lock, [&](){ return _loop != NULL; });
                loop = _loop;
            }
            return _loop;
        }
};


class EventLoopThreadPool {
    private:
        int _num_thread;
        int _next;
        EventLoop *_base_loop;
        std::vector<LoopThread*> _threads;
        std::vector<EventLoop*> _loops;
    public:
        EventLoopThreadPool::EventLoopThreadPool(EventLoop *baseloop):_base_loop(baseloop), _num_thread(0), _next(0) {}
        void EventLoopThreadPool::ser_thread_count(int num) {
            _num_thread = num;
        }
        EventLoop *EventLoopThreadPool::next_loop() {
            if (_num_thread == 0) {
                return _base_loop;
            }
            _next = (_next + 1) % _loops.size();
            return _loops[_next];
        }
        void EventLoopThreadPool::start() {
            if (_num_thread > 0) {
                _threads.resize(_num_thread);
                _loops.resize(_num_thread);
                for (int i = 0; i < _num_thread; i++) {
                    _threads[i] = new LoopThread();
                    _loops[i] = _threads[i]->own_loop();
                }
            }
            _base_loop->start();
        }
};